package com.flink.hbase.windowingoptimization.performance;

import com.flink.hbase.windowingoptimization.config.WatermarkConfig;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadLocalRandom;

/**
 * 窗口性能调优最佳实践指南
 * 
 * 优化策略：
 * 1. 窗口大小和滑动间隔优化
 * 2. 触发器和驱逐器优化
 * 3. 状态管理优化
 * 4. 内存使用优化
 * 5. 网络传输优化
 * 6. 并行度调优
 * 
 * 监控指标：
 * - 窗口处理延迟
 * - 内存使用量
 * - 吞吐量
 * - 反压情况
 * - 状态大小
 * 
 * @author Flink Developer
 */
public class PerformanceTuningGuide {
    
    private static final Logger logger = LoggerFactory.getLogger(PerformanceTuningGuide.class);
    
    /**
     * 性能测试事件
     */
    public static class PerformanceEvent {
        public String key;
        public double value;
        public long timestamp;
        public String category;
        public int size;
        public Map<String, Object> metadata;
        
        public PerformanceEvent() {
            this.metadata = new HashMap<>();
        }
        
        public PerformanceEvent(String key, double value, long timestamp, 
                              String category, int size) {
            this.key = key;
            this.value = value;
            this.timestamp = timestamp;
            this.category = category;
            this.size = size;
            this.metadata = new HashMap<>();
        }
        
        @Override
        public String toString() {
            return String.format("PerformanceEvent{key='%s', value=%.2f, timestamp=%d, " +
                    "category='%s', size=%d}",
                    key, value, timestamp, category, size);
        }
    }
    
    /**
     * 性能指标
     */
    public static class PerformanceMetrics {
        public long windowStartTime;
        public long windowEndTime;
        public long processingStartTime;
        public long processingEndTime;
        public long eventCount;
        public long totalSize;
        public double avgProcessingTime;
        public double throughput;
        public long memoryUsage;
        
        public PerformanceMetrics() {
            this.processingStartTime = System.currentTimeMillis();
        }
        
        public void finishProcessing() {
            this.processingEndTime = System.currentTimeMillis();
            this.avgProcessingTime = processingEndTime - processingStartTime;
            this.throughput = eventCount / (avgProcessingTime / 1000.0);
        }
        
        @Override
        public String toString() {
            return String.format("PerformanceMetrics{eventCount=%d, processingTime=%.2f ms, " +
                    "throughput=%.2f events/s, memoryUsage=%d bytes}",
                    eventCount, avgProcessingTime, throughput, memoryUsage);
        }
    }
    
    /**
     * 优化的聚合函数 - 使用增量聚合
     */
    public static class OptimizedAggregateFunction 
            implements AggregateFunction<PerformanceEvent, PerformanceAccumulator, PerformanceMetrics> {
        
        @Override
        public PerformanceAccumulator createAccumulator() {
            return new PerformanceAccumulator();
        }
        
        @Override
        public PerformanceAccumulator add(PerformanceEvent event, PerformanceAccumulator acc) {
            acc.count++;
            acc.sum += event.value;
            acc.totalSize += event.size;
            acc.minValue = Math.min(acc.minValue, event.value);
            acc.maxValue = Math.max(acc.maxValue, event.value);
            
            // 轻量级统计，避免存储所有事件
            if (acc.count == 1) {
                acc.firstEventTime = event.timestamp;
            }
            acc.lastEventTime = event.timestamp;
            
            return acc;
        }
        
        @Override
        public PerformanceMetrics getResult(PerformanceAccumulator acc) {
            PerformanceMetrics metrics = new PerformanceMetrics();
            metrics.eventCount = acc.count;
            metrics.totalSize = acc.totalSize;
            metrics.finishProcessing();
            
            // 估算内存使用
            metrics.memoryUsage = estimateMemoryUsage(acc);
            
            return metrics;
        }
        
        @Override
        public PerformanceAccumulator merge(PerformanceAccumulator acc1, PerformanceAccumulator acc2) {
            acc1.count += acc2.count;
            acc1.sum += acc2.sum;
            acc1.totalSize += acc2.totalSize;
            acc1.minValue = Math.min(acc1.minValue, acc2.minValue);
            acc1.maxValue = Math.max(acc1.maxValue, acc2.maxValue);
            acc1.firstEventTime = Math.min(acc1.firstEventTime, acc2.firstEventTime);
            acc1.lastEventTime = Math.max(acc1.lastEventTime, acc2.lastEventTime);
            
            return acc1;
        }
        
        private long estimateMemoryUsage(PerformanceAccumulator acc) {
            // 估算累加器内存使用
            return 8 * 6 + 16; // 6个long/double字段 + 对象头
        }
    }
    
    /**
     * 性能累加器
     */
    public static class PerformanceAccumulator {
        public long count = 0;
        public double sum = 0;
        public long totalSize = 0;
        public double minValue = Double.MAX_VALUE;
        public double maxValue = Double.MIN_VALUE;
        public long firstEventTime = Long.MAX_VALUE;
        public long lastEventTime = Long.MIN_VALUE;
    }
    
    /**
     * 优化的窗口函数 - 减少状态存储
     */
    public static class OptimizedWindowFunction 
            extends ProcessWindowFunction<PerformanceMetrics, PerformanceMetrics, String, TimeWindow> {
        
        private transient ValueState<PerformanceMetrics> previousMetrics;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            
            ValueStateDescriptor<PerformanceMetrics> descriptor = 
                new ValueStateDescriptor<>("previous-metrics", PerformanceMetrics.class);
            previousMetrics = getRuntimeContext().getState(descriptor);
        }
        
        @Override
        public void process(String key, Context context, 
                          Iterable<PerformanceMetrics> elements, 
                          Collector<PerformanceMetrics> out) throws Exception {
            
            PerformanceMetrics current = elements.iterator().next();
            current.windowStartTime = context.window().getStart();
            current.windowEndTime = context.window().getEnd();
            
            // 与上一个窗口对比
            PerformanceMetrics previous = previousMetrics.value();
            if (previous != null) {
                analyzePerformanceChange(current, previous);
            }
            
            // 更新状态
            previousMetrics.update(current);
            
            out.collect(current);
        }
        
        private void analyzePerformanceChange(PerformanceMetrics current, PerformanceMetrics previous) {
            double throughputChange = (current.throughput - previous.throughput) / previous.throughput;
            
            if (Math.abs(throughputChange) > 0.1) { // 10%变化
                logger.info("吞吐量变化 [{}]: {:.2f} -> {:.2f} ({:.2f}%)",
                    current.windowStartTime, previous.throughput, current.throughput, 
                    throughputChange * 100);
            }
            
            // 内存使用变化
            double memoryChange = (double)(current.memoryUsage - previous.memoryUsage) / previous.memoryUsage;
            if (Math.abs(memoryChange) > 0.2) { // 20%变化
                logger.warn("内存使用变化: {} -> {} ({:.2f}%)",
                    previous.memoryUsage, current.memoryUsage, memoryChange * 100);
            }
        }
    }
    
    /**
     * 高性能滑动窗口示例
     */
    public void runHighPerformanceSlidingWindow() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 性能调优配置
        env.setParallelism(8); // 根据CPU核心数调整
        env.setBufferTimeout(10); // 10ms缓冲超时，平衡延迟和吞吐量
        env.getConfig().setLatencyTrackingInterval(1000); // 1秒延迟跟踪
        
        // 数据源
        DataStream<PerformanceEvent> eventStream = env.addSource(new HighPerformanceEventSource());
        
        // 设置Watermark - 低延迟策略
        SingleOutputStreamOperator<PerformanceEvent> watermarkedStream = eventStream
            .assignTimestampsAndWatermarks(
                WatermarkConfig.lowLatencyStrategy(
                    (event, recordTimestamp) -> event.timestamp
                )
            );
        
        // 使用增量聚合的滑动窗口
        SingleOutputStreamOperator<PerformanceMetrics> windowedStream = watermarkedStream
            .keyBy(event -> event.category)
            .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
            .aggregate(
                new OptimizedAggregateFunction(),
                new OptimizedWindowFunction()
            )
            .name("high-performance-sliding-window");
        
        windowedStream.addSink(new PerformanceMetricsSink()).name("performance-sink");
        
        env.execute("High Performance Sliding Window");
    }
    
    /**
     * 内存优化的窗口处理
     */
    public void runMemoryOptimizedWindow() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        
        DataStream<PerformanceEvent> eventStream = env.addSource(new HighPerformanceEventSource());
        
        SingleOutputStreamOperator<PerformanceEvent> watermarkedStream = eventStream
            .assignTimestampsAndWatermarks(
                WatermarkConfig.mixedScenarioStrategy(
                    (event, recordTimestamp) -> event.timestamp
                )
            );
        
        // 使用自定义触发器减少内存使用
        SingleOutputStreamOperator<PerformanceMetrics> memoryOptimizedStream = watermarkedStream
            .keyBy(event -> event.category)
            .window(TumblingEventTimeWindows.of(Time.minutes(10)))
            .trigger(CountTrigger.of(1000)) // 每1000条记录触发一次
            .evictor(CountEvictor.of(5000)) // 保留最近5000条记录
            .aggregate(
                new OptimizedAggregateFunction(),
                new MemoryOptimizedWindowFunction()
            )
            .name("memory-optimized-window");
        
        memoryOptimizedStream.addSink(new PerformanceMetricsSink()).name("memory-optimized-sink");
        
        env.execute("Memory Optimized Window");
    }
    
    /**
     * 内存优化的窗口函数
     */
    public static class MemoryOptimizedWindowFunction 
            extends ProcessWindowFunction<PerformanceMetrics, PerformanceMetrics, String, TimeWindow> {
        
        @Override
        public void process(String key, Context context, 
                          Iterable<PerformanceMetrics> elements, 
                          Collector<PerformanceMetrics> out) throws Exception {
            
            PerformanceMetrics metrics = elements.iterator().next();
            metrics.windowStartTime = context.window().getStart();
            metrics.windowEndTime = context.window().getEnd();
            
            // 监控内存使用
            long usedMemory = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
            long maxMemory = Runtime.getRuntime().maxMemory();
            double memoryUsagePercent = (double) usedMemory / maxMemory * 100;
            
            if (memoryUsagePercent > 80) {
                logger.warn("内存使用率过高: {:.2f}% ({} / {})", 
                    memoryUsagePercent, usedMemory, maxMemory);
                
                // 触发垃圾回收
                System.gc();
            }
            
            metrics.memoryUsage = usedMemory;
            out.collect(metrics);
        }
    }
    
    /**
     * 并行度优化示例
     */
    public void runParallelismOptimization() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 动态并行度配置
        int sourceParallelism = 2;  // 数据源并行度
        int windowParallelism = 8;  // 窗口操作并行度
        int sinkParallelism = 4;    // 输出并行度
        
        DataStream<PerformanceEvent> eventStream = env
            .addSource(new HighPerformanceEventSource())
            .setParallelism(sourceParallelism);
        
        SingleOutputStreamOperator<PerformanceEvent> watermarkedStream = eventStream
            .assignTimestampsAndWatermarks(
                WatermarkConfig.highThroughputStrategy(
                    (event, recordTimestamp) -> event.timestamp
                )
            )
            .setParallelism(sourceParallelism);
        
        // 窗口操作使用更高并行度
        SingleOutputStreamOperator<PerformanceMetrics> windowedStream = watermarkedStream
            .keyBy(event -> event.category)
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .aggregate(
                new OptimizedAggregateFunction(),
                new ParallelismOptimizedWindowFunction()
            )
            .setParallelism(windowParallelism)
            .name("parallelism-optimized-window");
        
        // 输出操作使用中等并行度
        windowedStream
            .addSink(new PerformanceMetricsSink())
            .setParallelism(sinkParallelism)
            .name("parallelism-optimized-sink");
        
        env.execute("Parallelism Optimization");
    }
    
    /**
     * 并行度优化的窗口函数
     */
    public static class ParallelismOptimizedWindowFunction 
            extends ProcessWindowFunction<PerformanceMetrics, PerformanceMetrics, String, TimeWindow> {
        
        @Override
        public void process(String key, Context context, 
                          Iterable<PerformanceMetrics> elements, 
                          Collector<PerformanceMetrics> out) throws Exception {
            
            PerformanceMetrics metrics = elements.iterator().next();
            metrics.windowStartTime = context.window().getStart();
            metrics.windowEndTime = context.window().getEnd();
            
            // 记录并行度信息
            int parallelism = getRuntimeContext().getNumberOfParallelSubtasks();
            int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
            
            logger.debug("窗口处理 [{}]: 并行度={}, 子任务索引={}, 事件数={}",
                key, parallelism, subtaskIndex, metrics.eventCount);
            
            out.collect(metrics);
        }
    }
    
    /**
     * 网络优化示例
     */
    public void runNetworkOptimization() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 网络优化配置
        env.setParallelism(4);
        env.setBufferTimeout(50); // 50ms缓冲超时
        env.getConfig().setMaxParallelism(128); // 最大并行度
        
        DataStream<PerformanceEvent> eventStream = env.addSource(new HighPerformanceEventSource());
        
        SingleOutputStreamOperator<PerformanceEvent> watermarkedStream = eventStream
            .assignTimestampsAndWatermarks(
                WatermarkConfig.highThroughputStrategy(
                    (event, recordTimestamp) -> event.timestamp
                )
            );
        
        // 预聚合减少网络传输
        SingleOutputStreamOperator<PerformanceEvent> preAggregatedStream = watermarkedStream
            .keyBy(event -> event.category + "_" + (event.hashCode() % 4))
            .map(new PreAggregationFunction())
            .filter(Objects::nonNull)
            .name("pre-aggregation");
        
        // 主窗口聚合
        SingleOutputStreamOperator<PerformanceMetrics> windowedStream = preAggregatedStream
            .keyBy(event -> event.category)
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .aggregate(
                new OptimizedAggregateFunction(),
                new NetworkOptimizedWindowFunction()
            )
            .name("network-optimized-window");
        
        windowedStream.addSink(new PerformanceMetricsSink()).name("network-optimized-sink");
        
        env.execute("Network Optimization");
    }
    
    /**
     * 预聚合函数
     */
    public static class PreAggregationFunction extends RichMapFunction<PerformanceEvent, PerformanceEvent> {
        
        private transient ValueState<PerformanceEvent> localAggregateState;
        private transient ValueState<Integer> countState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            
            localAggregateState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("local-aggregate", PerformanceEvent.class));
            countState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("count", Integer.class, 0));
        }
        
        @Override
        public PerformanceEvent map(PerformanceEvent event) throws Exception {
            PerformanceEvent localAggregate = localAggregateState.value();
            Integer count = countState.value();
            
            if (localAggregate == null) {
                localAggregate = new PerformanceEvent(event.key, event.value, event.timestamp, 
                                                    event.category, event.size);
                count = 1;
            } else {
                localAggregate.value += event.value;
                localAggregate.size += event.size;
                localAggregate.timestamp = Math.max(localAggregate.timestamp, event.timestamp);
                count++;
            }
            
            localAggregateState.update(localAggregate);
            countState.update(count);
            
            // 每100个事件输出一次聚合结果
            if (count >= 100) {
                PerformanceEvent result = new PerformanceEvent(
                    localAggregate.key, localAggregate.value, localAggregate.timestamp,
                    localAggregate.category, localAggregate.size);
                
                // 重置状态
                localAggregateState.clear();
                countState.clear();
                
                return result;
            }
            
            return null;
        }
    }
    
    /**
     * 网络优化的窗口函数
     */
    public static class NetworkOptimizedWindowFunction 
            extends ProcessWindowFunction<PerformanceMetrics, PerformanceMetrics, String, TimeWindow> {
        
        @Override
        public void process(String key, Context context, 
                          Iterable<PerformanceMetrics> elements, 
                          Collector<PerformanceMetrics> out) throws Exception {
            
            PerformanceMetrics metrics = elements.iterator().next();
            metrics.windowStartTime = context.window().getStart();
            metrics.windowEndTime = context.window().getEnd();
            
            // 计算网络传输效率
            double dataReduction = calculateDataReduction(metrics);
            
            if (dataReduction > 0.5) { // 50%以上压缩率
                logger.info("网络传输优化效果: 数据压缩率 {:.2f}%", dataReduction * 100);
            }
            
            out.collect(metrics);
        }
        
        private double calculateDataReduction(PerformanceMetrics metrics) {
            // 简化的数据压缩率计算
            long originalSize = metrics.eventCount * 100; // 假设每个事件100字节
            long compressedSize = metrics.totalSize;
            
            return (double) (originalSize - compressedSize) / originalSize;
        }
    }
    
    /**
     * 高性能事件源
     */
    public static class HighPerformanceEventSource implements SourceFunction<PerformanceEvent> {
        
        private volatile boolean running = true;
        private final ThreadLocalRandom random = ThreadLocalRandom.current();
        
        @Override
        public void run(SourceContext<PerformanceEvent> ctx) throws Exception {
            String[] categories = {"A", "B", "C", "D", "E"};
            long eventId = 0;
            
            while (running) {
                // 批量生成事件减少锁竞争
                List<PerformanceEvent> batch = new ArrayList<>();
                
                for (int i = 0; i < 100; i++) {
                    PerformanceEvent event = new PerformanceEvent(
                        "key_" + (eventId % 1000),
                        random.nextDouble() * 100,
                        System.currentTimeMillis(),
                        categories[random.nextInt(categories.length)],
                        random.nextInt(1000) + 100
                    );
                    
                    batch.add(event);
                    eventId++;
                }
                
                // 批量输出
                for (PerformanceEvent event : batch) {
                    ctx.collect(event);
                }
                
                // 控制生成速率
                Thread.sleep(1);
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 性能指标Sink
     */
    public static class PerformanceMetricsSink implements SinkFunction<PerformanceMetrics> {
        
        private final Map<String, Long> categoryMetrics = new ConcurrentHashMap<>();
        private long totalEvents = 0;
        private long lastReportTime = System.currentTimeMillis();
        
        @Override
        public void invoke(PerformanceMetrics metrics, Context context) throws Exception {
            totalEvents += metrics.eventCount;
            
            // 定期输出性能报告
            long currentTime = System.currentTimeMillis();
            if (currentTime - lastReportTime > 30000) { // 30秒
                reportPerformance();
                lastReportTime = currentTime;
            }
            
            // 检查性能异常
            if (metrics.avgProcessingTime > 1000) { // 超过1秒
                logger.warn("窗口处理时间过长: {:.2f}ms, 事件数: {}", 
                    metrics.avgProcessingTime, metrics.eventCount);
            }
            
            if (metrics.throughput < 100) { // 低于100 events/s
                logger.warn("吞吐量过低: {:.2f} events/s", metrics.throughput);
            }
        }
        
        private void reportPerformance() {
            logger.info("=== 性能报告 ===");
            logger.info("总处理事件数: {}", totalEvents);
            logger.info("平均处理速率: {:.2f} events/s", 
                totalEvents / ((System.currentTimeMillis() - lastReportTime) / 1000.0));
            
            // 内存使用情况
            Runtime runtime = Runtime.getRuntime();
            long usedMemory = runtime.totalMemory() - runtime.freeMemory();
            logger.info("内存使用: {} MB / {} MB", 
                usedMemory / 1024 / 1024, runtime.maxMemory() / 1024 / 1024);
            
            logger.info("================");
        }
    }
    
    /**
     * 性能测试工具
     */
    public void runPerformanceTest() throws Exception {
        logger.info("开始性能测试...");
        
        // 测试不同配置的性能
        String[] testCases = {
            "high-performance-sliding",
            "memory-optimized",
            "parallelism-optimized",
            "network-optimized"
        };
        
        for (String testCase : testCases) {
            logger.info("执行测试案例: {}", testCase);
            
            long startTime = System.currentTimeMillis();
            
            try {
                switch (testCase) {
                    case "high-performance-sliding":
                        runHighPerformanceSlidingWindow();
                        break;
                    case "memory-optimized":
                        runMemoryOptimizedWindow();
                        break;
                    case "parallelism-optimized":
                        runParallelismOptimization();
                        break;
                    case "network-optimized":
                        runNetworkOptimization();
                        break;
                }
            } catch (Exception e) {
                logger.error("测试案例 {} 执行失败", testCase, e);
            }
            
            long endTime = System.currentTimeMillis();
            logger.info("测试案例 {} 完成，耗时: {}ms", testCase, endTime - startTime);
        }
        
        logger.info("性能测试完成");
    }
    
    public static void main(String[] args) throws Exception {
        PerformanceTuningGuide guide = new PerformanceTuningGuide();
        
        String mode = args.length > 0 ? args[0] : "high-performance";
        
        switch (mode) {
            case "high-performance":
                guide.runHighPerformanceSlidingWindow();
                break;
            case "memory-optimized":
                guide.runMemoryOptimizedWindow();
                break;
            case "parallelism-optimized":
                guide.runParallelismOptimization();
                break;
            case "network-optimized":
                guide.runNetworkOptimization();
                break;
            case "performance-test":
                guide.runPerformanceTest();
                break;
            default:
                System.out.println("支持的模式: high-performance, memory-optimized, " +
                                 "parallelism-optimized, network-optimized, performance-test");
                break;
        }
    }
} 